package org.example;


import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.TopicDescription;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * @Desc:
 * @Author: yanghongjun
 * @Date: 2024/4/24
 **/
public class KafkaTopicViewer {

    private static final String BOOTSTRAP_SERVERS = "192.168.0.194:9092"; // 替换为你的 Kafka 服务器地址

    public static void main(String[] args) {
        Properties adminProps = new Properties();
        adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);

        try (AdminClient adminClient = AdminClient.create(adminProps)) {
            // 查看所有 Topic 列表
            viewAllTopics(adminClient);

            // 查看特定 Topic 详细信息（此处以 "myTopic" 为例）
            //viewTopicDetails(adminClient, "myTopic");
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    private static void viewAllTopics(AdminClient adminClient) throws InterruptedException, ExecutionException {
        ListTopicsResult topicsResult = adminClient.listTopics();

        System.out.println("All Topics:");
        for (String topic : topicsResult.names().get()) {
            System.out.println("- " + topic);
        }
    }

    private static void viewTopicDetails(AdminClient adminClient, String topicName) throws InterruptedException, ExecutionException {
        TopicDescription topicDescription = adminClient.describeTopics(Collections.singletonList(topicName)).values().get(topicName).get();

        System.out.println("\nTopic Details for '" + topicName + "':");
        System.out.println("Partition Count: " + topicDescription.partitions().size());
        System.out.println("Replication Factor: " + topicDescription.partitions().iterator().next().replicas().size());

        topicDescription.partitions().forEach(partitionInfo -> {
            System.out.println("\nPartition ID: " + partitionInfo.partition());
            System.out.println("Leader: " + partitionInfo.leader().id());
            System.out.println("Replicas: " + partitionInfo.replicas().stream().map(b -> b.id() + "").collect(Collectors.joining(", ")));
            System.out.println("In-Sync Replicas: " + partitionInfo.isr().stream().map(b -> b.id() + "").collect(Collectors.joining(", ")));
        });
    }
}